package 多线程练习;

import java.util.Random;
import java.util.concurrent.*;

// 函数式接口
@FunctionalInterface
interface TaskProcessor<T> {
    void process(T task) throws InterruptedException;
}

// 泛型任务类
class Task<T> {
    private final T data;

    public Task(T data) {
        this.data = data;
    }

    public T getData() {
        return data;
    }
    // 构造函数、getter
}

// 任务处理器工厂（使用静态方法返回lambda表达式）
class ProcessorFactory {
    static TaskProcessor<String> createStringProcessor() {
        return task-> {
            System.out.println(Thread.currentThread().getName()+"现在处理的是字符类型 ,这是处理结果："+task.toUpperCase());
            Thread.sleep(80);
        };
    }
    static TaskProcessor<Integer> createNumberProcessor() {
        return task->{
            System.out.println(Thread.currentThread().getName()+"现在处理的是整形类型，这是处理结果："+task*task);
            Thread.sleep(80);
        };
    }
}

// 主系统类
public class TaskProcessingSystem {
    // 使用 ThreadPoolExecutor 管理线程
    // 实现生产者-消费者模式
    // 添加优雅关闭机制
    private final int QUEUE_CAPACITY=10;
    private final BlockingQueue<Task<?>> queue=new LinkedBlockingQueue<>(QUEUE_CAPACITY);
    private final ExecutorService executor= Executors.newFixedThreadPool(5);
    private boolean running=true;
    public void start(){
        for(int i=1;i<=3;i++){
            final int pid=i;
            executor.submit(()->{
                try{
                    produceTasks(pid);
                }catch (InterruptedException e){
                    Thread.currentThread().interrupt();
                }
            });
        }
        for(int i=0;i<2;i++){
            executor.submit(()->{
                try{
                    consumeTasks();
                }catch (InterruptedException e){
                    Thread.currentThread().interrupt();
                }
            });
        }
    }
    public void produceTasks(int pid) throws InterruptedException {
        Random random=new Random();
        Task<?> task;
        String[] samples={"Apple","Banana","Orange","END"};
        for(String sample:samples){
            if ("END".equals(sample) && pid==3){
                task=new Task<>(sample);
            }else{
                task=(pid%2==0)?
                        new Task<>(random.nextInt(50)):
                        new Task<>(sample+"-"+pid);
            }
            queue.put(task);
            System.out.printf("produceId :%d 添加%s%n",pid,task.getData());
            Thread.sleep(random.nextInt(200));
        }
    }
    public void consumeTasks() throws InterruptedException{
        TaskProcessor<String> StringProcessor=ProcessorFactory.createStringProcessor();
        TaskProcessor<Integer> IntegerProcessor=ProcessorFactory.createNumberProcessor();
        while(running || !Thread.currentThread().isInterrupted()){
            Task<?> task=queue.take();
            Object data=task.getData();
            if("END".equals(data)){
                shutdown();
                return;
            }
            if(data instanceof String){
                StringProcessor.process((String) data);
            }else if(data instanceof Integer){
                IntegerProcessor.process((Integer) data);
            }
        }
    }
    public void shutdown(){
        running=false;
        executor.shutdownNow();
        try{
            if(!executor.awaitTermination(800,TimeUnit.MILLISECONDS)){
                System.err.println("强制关闭剩余线程");
            }
        }catch (InterruptedException e){
            Thread.currentThread().interrupt();
        }
        System.out.println("执行完毕");
    }

    public static void main(String[] args) {
        TaskProcessingSystem system=new TaskProcessingSystem();
        system.start();

    }
}
